package com.hadoop.hdfs;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.IOUtils;
import org.junit.Test;

public class HdfsApiDemo {
	public static String NAME_NODE_URL = "hdfs://192.168.186.6:9000";

	/*
	 * 测试连接
	 */
	@Test
	public void testConnect() throws Exception {
		// 可以通过Configuration对的Hadoop的环境进行配置
		Configuration conf = new Configuration();
		// 代码的设置的优先级高于文件的配置。（因为文件的配置是全局通用的)
		// conf.set("dfs.replication", "1");

		// Hadoop的文件系统有很多种，分布式文件系统是其中的一种，这也是最重要，最常用的文件系统。对应的类：
		// DistributedFileSystem。此外还有比如 webHDFS文件系统，本地文件系统等。
		FileSystem fs = FileSystem.get(new URI(NAME_NODE_URL), conf);

	}
	
	/*
	 * 将本地文件上传到HDFS
	 */
	@Test
	public void testPut() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new URI(NAME_NODE_URL), conf);
		// 获取本地文件的输入流
		InputStream in = new FileInputStream("5.txt");
		// 获取HDFS的输出流,注意path路径写到文件路径级别
		OutputStream out = fs.create(new Path("/park01/5.txt"));
		// 注意导包，是hadoop的包
		IOUtils.copyBytes(in, out, conf);

	}

	/*
	 * 将HDFS上的文件下载到本地
	 */
	@Test
	public void testGet() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new URI(NAME_NODE_URL), conf);
		// 获取HDFS指定的文件的输入流
		InputStream in = fs.open(new Path("/park02/2.txt"));

		OutputStream out = new FileOutputStream("2.txt");

		IOUtils.copyBytes(in, out, conf);
	}

	/*
	 * 删除HDFS上的文件或目录 目录重命名
	 */
	@Test
	public void testDelete_Rename() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new URI(NAME_NODE_URL), conf);
		// fs.delete(new Path("/park04"));
		// true表示递归删除，可以删除非空目录
		// fs.delete(new Path("/park02"),true );
		fs.rename(new Path("/park01"), new Path("/park04"));
		// 建议关闭
		fs.close();

	}

	/*
	 * 递归查看指定目录下的文件数据
	 */
	@Test
	public void testListFiles() throws IOException, URISyntaxException {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new URI(NAME_NODE_URL), conf);
		RemoteIterator<LocatedFileStatus> rt = fs.listFiles(new Path("/"), true);
		while (rt.hasNext()) {
			System.out.println(rt.next());
		}

	}

	@Test
	public void testGetBlockLocation() throws Exception {
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(new URI(NAME_NODE_URL), conf);
		// 如果想拿到所有块信息，len=Integer.MaxValue。

		BlockLocation[] bls = fs.getFileBlockLocations(new Path("/park02/hadoop"), 0, Integer.MAX_VALUE);
		for (BlockLocation bl : bls) {
			System.out.println(bl);
		}
	}
}
